package com.ziroom.crm.mq.core.persistence;

import cn.hutool.core.io.FileUtil;
import com.alibaba.fastjson.JSON;
import com.ziroom.crm.mq.core.config.MqConfig;
import com.ziroom.crm.mq.core.data.Message;
import com.ziroom.crm.mq.core.data.ProcessTypes;
import com.ziroom.crm.mq.core.exception.MqConsumeException;
import com.ziroom.crm.mq.core.util.FileUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.io.*;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * 保存文件
 * <p>
 * <PRE>
 * <BR>    修改记录
 * <BR>-----------------------------------------------
 * <BR>    修改日期         修改人          修改内容
 * </PRE>
 *
 * @author WeiShaoying
 * @version 1.0
 * @date Created in 2020年09月01日 14:25
 * @since 1.0
 */
@Slf4j
@SuppressWarnings("ALL")
public abstract class FilePersistence<T extends Message> implements Persistence<T> {

    /**
     * 换行分隔符
     */
    private static final String LINE_SEPARATOR = "line.separator";

    /**
     * 存储文件名
     */
    private static final String DEFAULT_FILE_NAME = "crm-mq";

    /**
     * 加锁
     */
    private static final Object LOCK = new Object();

    private static final Object READ_LOCK = new Object();

    private static final Object CLEAR_LOCK = new Object();


    @Override
    public String getType() {
        return Persistence.FILE;
    }

    @Override
    public void save(Message message, ProcessTypes processType) {
        if (Objects.isNull(message)) {
            return;
        }
        save(toString(message), processType);
    }

    /**
     * 保存消息
     *
     * @param msg         消息
     * @param processType 处理类型
     */
    private void save(String msg, ProcessTypes processType) {
        synchronized (LOCK) {
            Path path = getFilePath(processType);
            try {
                Files.write(path, msg.getBytes(), StandardOpenOption.APPEND);
            } catch (IOException e) {
                log.error("Write File fail", e);
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public void batchSave(List messageList, ProcessTypes processType) {
        if (Objects.isNull(messageList)) {
            return;
        }
        Path path = getFilePath(processType);
        List<String> messageStrList = new ArrayList<>();
        for (Object item : messageList) {
            // 序列化可自定义
            messageStrList.add(JSON.toJSONString(item));
        }
        try {
            Files.write(path, messageList, StandardOpenOption.APPEND);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public List queryMessage(int size, ProcessTypes processType, Date queryEndTime) {
        if (size <= 0) {
            throw new IllegalArgumentException("size do not allow less than zero");
        }
        synchronized (LOCK) {
            Path path = getFilePath(processType);
            Path tempFilePath = getTempFilePath(path);
            File inputFile = path.toFile();
            File tempFile = tempFilePath.toFile();

            List<T> messageList = new ArrayList<>(size);
            try {
                BufferedReader reader = new BufferedReader(new FileReader(inputFile));
                // 写到临时文件
                BufferedWriter writer = new BufferedWriter(new FileWriter(tempFile));
                String currentLine;
                int i = 0;

                while ((currentLine = reader.readLine()) != null) {
                    // 未匹配数据,留待下次处理
                    boolean match = false;
                    String line = currentLine.trim();
                    if (StringUtils.isBlank(line)) {
                        continue;
                    }
                    if (i < size) {
                        T object = deserialization(line);
                        if (Objects.nonNull(object)) {
                            Date nextRetryDate = object.getNextRetryDate();
                            // 已经达到重试时间
                            if (Objects.isNull(nextRetryDate) || nextRetryDate.before(queryEndTime)) {
                                // 超过查询时间，跳出
                                messageList.add(object);
                                match = true;
                                i++;
                            }
                        } else {
                            // 反序列化失败=>死信
                            ProcessTypes deadType = processType.getDeadType();
                            if (Objects.nonNull(deadType)) {
                                save(addLineSeparator(line), deadType);
                            }
                        }
                    }

                    if (!match) {
                        //写其他行,查询的行直接删除
                        writer.write(currentLine + System.getProperty(LINE_SEPARATOR));
                    }
                }

                writer.close();
                reader.close();

                if (i == 0) {
                    // 直接返回
                    return messageList;
                }
                // 把临时文件重命名
                Files.move(tempFilePath, path, StandardCopyOption.ATOMIC_MOVE);
            } catch (IOException e) {
                log.error("create temp file", e);
                throw new MqConsumeException(e);
            }
            return messageList;
        }
    }

    /**
     * 查询所有消息
     *
     * @param processTypes
     * @return
     */
    @Override
    public List<T> queryAllMessage(ProcessTypes... processTypes) {
        synchronized (READ_LOCK) {
            List<T> messageList = new ArrayList<>();

            for (ProcessTypes processType : processTypes) {
                Path path = getFilePath(processType);
                File file = path.toFile();

                BufferedReader reader = null;
                try {
                    reader = new BufferedReader(new FileReader(file));
                    String currentLine;

                    while ((currentLine = reader.readLine()) != null) {
                        String line = currentLine.trim();
                        if (StringUtils.isBlank(line)) {
                            continue;
                        }
                        T message = deserialization(line);
                        if (Objects.nonNull(message)) {
                            messageList.add(message);
                        }
                    }
                } catch (Exception e) {
                    log.error("queryAllMessage fail", e);
                } finally {
                    if (reader != null) {
                        try {
                            reader.close();
                        } catch (IOException e) {
                            log.error("Close Reader fail", e);
                        }
                    }
                }
            }
            return messageList;
        }
    }

    @Override
    public void clear(ProcessTypes... processTypes) {
        synchronized (CLEAR_LOCK) {
            for (ProcessTypes processType : processTypes) {
                Path path = getFilePath(processType);
                File file = path.toFile();
                clearFile(file);
                // 临时文件
                Path tempFilePath = getTempFilePath(path);
                File tempFile = tempFilePath.toFile();
                if (tempFile.exists()) {
                    clearFile(tempFile);
                }
            }
            log.info("====Clear Success");
        }
    }

    /**
     * 清空内容
     *
     * @param file
     */
    private void clearFile(File file) {
        BufferedWriter writer = null;
        try {
            writer = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(file, false), StandardCharsets.UTF_8));
            writer.write("");
            writer.flush();
        } catch (Exception e) {
            log.error("Clear File fail", e);
        } finally {
            if (writer != null) {
                try {
                    writer.close();
                } catch (IOException e) {
                    log.error("Close File fail", e);
                }
            }
        }
    }

    @Override
    public int getDataCount(ProcessTypes... processTypes) {
        int total = 0;
        for (ProcessTypes processType : processTypes) {
            Path path = getFilePath(processType);
            File file = path.toFile();
            total += Math.max(FileUtil.getTotalLines(file) - 1, 0);
        }
        log.info("====Count={}", total);
        return total;
    }

    /**
     * 实体转为String
     *
     * @param message
     * @return
     */
    private String toString(Message message) {
        return addLineSeparator(JSON.toJSONString(message));
    }

    /**
     * 添加换行符
     *
     * @param msg
     * @return
     */
    private String addLineSeparator(String msg) {
        return msg + System.getProperty(LINE_SEPARATOR);
    }

    /**
     * 反序列化
     *
     * @param str
     * @return
     */
    protected T deserialization(String str) {
        try {
            Type types = this.getClass().getGenericSuperclass();
            Type[] params = ((ParameterizedType) types).getActualTypeArguments();
            Class<T> clazz = (Class) params[0];
            T t = JSON.parseObject(str, clazz);
            return t;
        } catch (Exception e) {
            log.error("反序列化失败", e);
        }
        return null;
    }

    /**
     * 获取文件
     *
     * @return
     */
    protected Path getFilePath(ProcessTypes processType) {
        String filePath = MqConfig.getConfig(MqConfig.PERSISTENCE_FILE_PATH);

        Path path = Paths.get(filePath, DEFAULT_FILE_NAME + "_" + processType.name());
        FileUtils.createIfNoExist(path);

        if (log.isDebugEnabled()) {
            log.debug("storage message file path:" + path.toFile().getAbsoluteFile());
        }
        return path;
    }

    /**
     * 创建同目录文件
     *
     * @param path
     * @return
     */
    protected Path getTempFilePath(Path path) {
        Path tempPath = Paths.get(path.getParent().toString(), path.toFile().getName() + ".temp");
        FileUtils.createIfNoExist(tempPath);
        return tempPath;
    }

}
